/*!
 * \file ParserXeleSkt.cpp
 * \project	WonderTrader
 *
 * \author Wesley
 * \date 2020/03/30
 * 
 * \brief Xele Socket行情解析器实现文件
 * 
 * \details 该文件实现了ParserXeleSkt类的所有功能，主要包括：
 *          - Xele行情协议的完整解析和处理
 *          - TCP+UDP双协议网络架构的实现
 *          - 上海期货交易所(SHFE) Xele系统的专用适配
 *          - UDP组播实时行情数据接收和处理
 *          - TCP快照数据预加载和缓存管理
 *          - 多种Xele数据格式的解析和转换
 *          - 价格精度处理和数据标准化
 *          - 时间戳转换和毫秒级精度处理
 *          - 五档深度行情数据的完整支持
 *          - 异步网络编程和错误恢复机制
 *          - 与WonderTrader数据结构的无缝集成
 * 
 * \note 该实现专门针对SHFE的Xele行情系统，具有极低的延迟特性
 * \warning Xele协议的时间戳处理需要特别注意时区和精度问题
 */
#include "ParserXeleSkt.h"
#include "md_struct.h"
#include "../Includes/WTSVariant.hpp"
#include "../Includes/WTSDataDef.hpp"
#include "../Includes/IBaseDataMgr.h"
#include "../Includes/WTSContractInfo.hpp"

#include <boost/bind.hpp>

/*!
 * \brief 通用日志记录模板函数
 * \tparam Args 可变参数类型
 * \param sink 行情回调接口指针
 * \param ll 日志级别
 * \param format 格式字符串
 * \param args 可变参数列表
 * 
 * \details 提供格式化日志输出功能，支持：
 *          - 多种日志级别（INFO、ERROR、WARN、DEBUG）
 *          - 线程安全的格式化输出
 *          - 自动内存管理和缓冲区保护
 *          - 与WonderTrader日志系统集成
 * 
 * \note 使用fmtlib提供高性能格式化，线程本地存储确保安全性
 */
 //By Wesley @ 2022.01.05
#include "../Share/fmtlib.h"
template<typename... Args>
inline void write_log(IParserSpi* sink, WTSLogLevel ll, const char* format, const Args&... args)
{
	if (sink == NULL)
		return;

	static thread_local char buffer[512] = { 0 };
	memset(buffer, 0, 512);
	fmt::format_to(buffer, format, args...);

	sink->handleParserLog(ll, buffer);
}

/*!
 * \brief 时间字符串转换为数字时间
 * \param strTime 时间字符串（格式：HH:MM:SS）
 * \return 数字时间（格式：HHMMSS）
 * 
 * \details 将Xele系统的时间字符串转换为WonderTrader使用的数字时间格式：
 *          - 移除时间字符串中的冒号分隔符
 *          - 转换为6位数字格式（HHMMSS）
 *          - 支持标准的时间格式解析
 * 
 * \note 该函数专门用于处理Xele系统的时间格式
 */
inline uint32_t strToTime(const char* strTime)
{
	static char str[10] = { 0 };
	const char *pos = strTime;
	int idx = 0;
	auto len = strlen(strTime);
	for (std::size_t i = 0; i < len; i++)
	{
		if (strTime[i] != ':')
		{
			str[idx] = strTime[i];
			idx++;
		}
	}
	str[idx] = '\0';

	return strtoul(str, NULL, 10);
}

/*!
 * \brief C接口导出函数
 * 
 * 提供标准的动态库导出接口，支持插件化架构
 */
extern "C"
{
	/*!
	 * \brief 创建Xele行情解析器实例
	 * \return 新创建的行情解析器接口指针
	 * 
	 * 用于动态库插件系统，创建ParserXeleSkt实例
	 */
	EXPORT_FLAG IParserApi* createParser()
	{
		ParserXeleSkt* parser = new ParserXeleSkt();
		return parser;
	}

	/*!
	 * \brief 销毁Xele行情解析器实例
	 * \param parser 要销毁的行情解析器接口指针（引用传递，销毁后置为NULL）
	 * 
	 * 安全地销毁ParserXeleSkt实例并清理资源
	 */
	EXPORT_FLAG void deleteParser(IParserApi* &parser)
	{
		if (NULL != parser)
		{
			delete parser;
			parser = NULL;
		}
	}
};

/*!
 * \brief 构造函数
 * 
 * 初始化ParserXeleSkt实例的各个成员变量为默认值
 */
ParserXeleSkt::ParserXeleSkt()
	: _udp_socket(NULL)
	, _strand(_io_service)
	, _stopped(false)
	, _sink(NULL)
	, _prepared(false)
	, _tick_cache(NULL)
{
}

/*!
 * \brief 析构函数
 * 
 * 清理ParserXeleSkt实例，释放相关资源
 */
ParserXeleSkt::~ParserXeleSkt()
{
}

/*!
 * \brief 初始化Xele行情解析器
 * \param config 配置参数对象
 * \return 是否初始化成功
 * 
 * \details 从配置文件中读取Xele连接参数并初始化网络组件：
 *          - TCP连接配置：tcp_host（TCP服务器地址）、tcp_port（TCP端口）
 *          - UDP组播配置：mcast_host（组播地址）、mcast_port（组播端口）
 *          - 网络接口配置：local_host（本地网卡地址，用于组播绑定）
 *          - 性能参数：gpsize（分组大小，默认1000）
 *          - 网络地址解析：将主机地址解析为IP地址
 *          - 终端点配置：设置TCP和UDP终端点
 *          - 组播地址绑定：配置组播接收的本地绑定地址
 * 
 * \note Xele系统需要配置TCP和UDP两个不同的连接参数
 * \warning 必须确保组播地址和本地网卡地址配置正确
 */
bool ParserXeleSkt::init( WTSVariant* config )
{
	_tcp_host = config->getCString("tcp_host");
	_tcp_port = config->getInt32("tcp_port");
	_mcast_host = config->getCString("mcast_host");
	_mcast_port = config->getInt32("mcast_port");
	_local_host = config->getCString("local_host");
	_gpsize = config->getUInt32("gpsize");
	if (_gpsize == 0)
		_gpsize = 1000;

	ip::address addr = ip::make_address(_tcp_host);
	_tcp_ep = ip::tcp::endpoint(addr, _tcp_port);

	addr = ip::make_address("0.0.0.0");
	_mcast_ep = ip::udp::endpoint(addr, _mcast_port);

	return true;
}

/*!
 * \brief 释放资源
 * 
 * \details 释放Xele行情解析器占用的所有资源，当前实现为空
 * 
 * \note 资源清理主要通过disconnect()实现
 */
void ParserXeleSkt::release()
{
	
}

/*!
 * \brief 重新连接Xele服务器
 * \return 是否重连成功
 * 
 * \details 重新建立UDP组播连接，包括快照准备和组播接收：
 *          - 快照准备：如果未准备过，先通过TCP获取市场快照
 *          - UDP组播设置：
 *            - 关闭并重建UDP套接字
 *            - 设置地址重用选项
 *            - 绑定到组播端口
 *            - 加入组播组（指定本地网卡）
 *            - 启动异步接收
 * 
 * \note 使用boost::bind绑定异步回调函数
 * \warning 组播加入需要正确的本地网卡地址
 */
bool ParserXeleSkt::reconnect()
{
	if(!_prepared)
	{
		prepare();
		_prepared = true;
	}

	{// 设置组播通道
		if (_udp_socket != NULL)
		{
			_udp_socket->close();
			delete _udp_socket;
			_udp_socket = NULL;
		}

		_udp_socket = new ip::udp::socket(_io_service);

		_udp_socket->open(_mcast_ep.protocol());
		_udp_socket->set_option(ip::udp::socket::reuse_address(true));
		_udp_socket->bind(_mcast_ep);
		_udp_socket->set_option(ip::multicast::join_group(ip::make_address_v4(_mcast_host.c_str()), ip::make_address_v4(_local_host.c_str())));

		_udp_socket->async_receive_from(buffer(_udp_buffer), _udp_ep,
			boost::bind(&ParserXeleSkt::handle_udp_read, this,
			boost::asio::placeholders::error,
			boost::asio::placeholders::bytes_transferred));
	}
	write_log(_sink, LL_INFO, "[ParserXeleSkt] Ready to receive from multicast tunnel {}:{}...", _mcast_host, _mcast_port);
	return true;
}

/*!
 * \brief 准备市场快照数据
 * \return 是否准备成功
 * 
 * \details 通过TCP连接获取市场快照数据，初始化价格精度和缓存：
 *          - TCP连接建立：连接到Xele快照服务器
 *          - 数据接收：读取所有可用的快照数据
 *          - 数据解析：解析Xele快照数据包格式
 *          - 缓存初始化：创建Tick数据缓存
 *          - 价格精度设置：根据合约信息设置价格刻度
 *          - 数据转换：将Xele数据转换为WonderTrader格式
 *          - 缓存存储：将快照数据存储到缓存中
 * 
 * \note 快照数据用于初始化市场状态，确保实时数据的完整性
 * \warning TCP连接失败会影响数据的完整性
 */
bool ParserXeleSkt::prepare()
{
	boost::asio::io_context io_context;
	ip::tcp::socket s(io_context);
	boost::system::error_code ec;
	s.connect(_tcp_ep, ec);
	if(ec)
	{
		write_log(_sink, LL_ERROR, "[ParserXeleSkt] Preparing snapshots failed, connecting tcp tunnel {}:{} error: {}", _tcp_host, _tcp_port, ec.message());
		return false;
	}

	if (_tick_cache == NULL)
		_tick_cache = TickCache::create();
	
	write_log(_sink, LL_INFO, "[ParserXeleSkt] Preparing snapshots via tcp tunnel {}:{}...", _tcp_host, _tcp_port);

	boost::array<char, 4096> buffer;
	std::string content;
	auto snap_size = sizeof(CXeleShfeSnapShot) + sizeof(CXeleShfeMarketHead);
	for (;;)
	{
		std::size_t reply_length = boost::asio::read(s, boost::asio::buffer(buffer), ec);
		if (ec)
		{
			// 接收错误
			break;
		}

		if (reply_length > 0)
			content.append(buffer.data(), reply_length);
	}

	while (content.size() > snap_size)
	{
		CXeleShfeMarketHead *mh = (CXeleShfeMarketHead *)(content.data());
		int8_t version = mh->Version;
		if (version != XELE_MD_DATA_VERSION)
			break;

		int8_t type = mh->Type;
		if (type != MESSAGE_SNAP_SHOT)
			break;

		CXeleShfeSnapShot *p = (CXeleShfeSnapShot *)(content.data() + sizeof(CXeleShfeMarketHead));
		int instrumentNo = p->InstrumentNo;
		write_log(_sink, LL_DEBUG, "[ParserXeleSkt] {}-{}", p->InstrumentID, instrumentNo);
		WTSContractInfo* ct = _bd_mgr->getContract(p->InstrumentID);
		if (ct != NULL)
		{
			_price_scales[instrumentNo] = p->PriceTick;

			WTSTickData* tick = WTSTickData::create(p->InstrumentID);
			tick->setContractInfo(ct);

			WTSTickStruct& quote = tick->getTickStruct();
			wt_strcpy(quote.exchg, ct->getExchg());

			quote.action_date = strToTime(p->ActionDay);
			quote.action_time = strToTime(p->UpdateTime) * 1000 + p->UpdateMilliSec;

			quote.price = p->LastPrice;
			quote.open = p->OpenPrice;
			quote.high = p->HighestPrice;
			quote.low = p->LowestPrice;
			quote.total_volume = p->Volume;
			quote.trading_date = quote.action_date;
			quote.settle_price = p->SettlementPrice;
			quote.total_turnover = p->Turnover;

			quote.open_interest = p->OpenInterest;

			quote.upper_limit = p->UpperLimitPrice;
			quote.lower_limit = p->LowerLimitPrice;

			quote.pre_close = p->PreClosePrice;
			quote.pre_settle = p->PreSettlementPrice;
			quote.pre_interest = p->PreOpenInterest;

			// 委托价格
			quote.ask_prices[0] = p->AskPrice;
			// 委托价格
			quote.bid_prices[0] = p->BidPrice;
			// 委托量
			quote.ask_qty[0] = p->AskVolume;
			// 委托量
			quote.bid_qty[0] = p->BidVolume;

			_tick_cache->add(instrumentNo, tick, false);

			if (_sink)
				_sink->handleQuote(tick, 1);
		}

		content.erase(0, snap_size);
	}
	
	s.close(ec);

	write_log(_sink, LL_INFO, "[ParserXeleSkt] All snapshots synced");

	return true;
}

/*!
 * \brief 建立连接
 * \return 是否连接成功
 * 
 * \details 启动Xele网络连接：
 *          - 调用reconnect()建立UDP组播连接
 *          - 运行io_service事件循环
 *          - 开始异步网络事件处理
 * 
 * \note 网络处理在当前线程中运行，会阻塞直到连接断开
 */
bool ParserXeleSkt::connect()
{
	if(reconnect())
	{
		_io_service.run();
	}
	else
	{
		return false;
	}

	return true;
}

/*!
 * \brief 断开连接
 * \return 是否断开成功
 * 
 * \details 安全断开Xele连接：
 *          - 关闭并删除UDP套接字
 *          - 设置停止标志
 *          - 通过strand投递断开事件处理
 *          - 确保线程安全的断开流程
 * 
 * \note 使用strand确保断开操作的线程安全性
 */
bool ParserXeleSkt::disconnect()
{
	if(_udp_socket != NULL)
	{
		_udp_socket->close();
		delete _udp_socket;
		_udp_socket = NULL;
	}

	_stopped = true;
	boost::asio::post(_strand, boost::bind(&ParserXeleSkt::doOnDisconnected, this));

	return true;
}

/*!
 * \brief 检查连接状态
 * \return 是否已连接
 * 
 * \details 通过检查UDP套接字是否有效来判断连接状态
 * 
 * \note 以UDP套接字为准判断连接状态
 */
bool ParserXeleSkt::isConnected()
{
	return _udp_socket!=NULL;
}

/*!
 * \brief 订阅行情数据
 * \param vecSymbols 要订阅的合约代码集合
 * 
 * \details 将合约代码添加到订阅集合：
 *          - 遍历传入的合约代码
 *          - 检查代码是否已存在于订阅集合中
 *          - 避免重复订阅同一合约
 *          - 将新代码加入订阅集合
 * 
 * \note 订阅集合用于过滤接收到的行情数据
 */
void ParserXeleSkt::subscribe( const CodeSet &vecSymbols )
{
	auto cit = vecSymbols.begin();
	for(; cit != vecSymbols.end(); cit++)
	{
		const auto &code = *cit;
		if(_set_subs.find(code) == _set_subs.end())
		{
			_set_subs.insert(code);
		}
	}
}

/*!
 * \brief 退订行情数据
 * \param setSymbols 要退订的合约代码集合
 * 
 * \details 处理行情退订请求，当前实现为空，可根据需要扩展
 * 
 * \note 预留接口，用于实现行情数据的动态退订功能
 */
void ParserXeleSkt::unsubscribe(const CodeSet &setSymbols)
{

}

/*!
 * \brief 注册行情回调接口
 * \param listener 行情数据回调接口指针
 * 
 * \details 注册行情数据处理接口：
 *          - 检查是否替换了已有的监听器
 *          - 保存新的回调接口指针
 *          - 记录监听器替换的调试信息
 *          - 获取基础数据管理器引用
 * 
 * \note 基础数据管理器用于获取合约信息
 */
void ParserXeleSkt::registerSpi( IParserSpi* listener )
{
	bool bReplaced = (_sink!=NULL);
	_sink = listener;
	if(bReplaced && _sink)
	{
		write_log(_sink, LL_DEBUG, "[ParserXeleSkt] Listener is replaced");
	}

	if (_sink)
		_bd_mgr = _sink->getBaseDataMgr();
}

/*!
 * \brief 处理UDP组播数据接收
 * \param e boost错误代码
 * \param bytes_transferred 接收到的字节数
 * 
 * \details 异步处理UDP组播数据接收：
 *          - 检查接收是否发生错误
 *          - 错误时触发断开事件和重连机制
 *          - 成功时调用extract_buffer解析数据
 *          - 继续启动下一次异步接收
 * 
 * \note 网络错误时会自动尝试重连（2秒延迟）
 * \warning 停止状态下不会进行重连和数据处理
 */
void ParserXeleSkt::handle_udp_read(const boost::system::error_code& e, std::size_t bytes_transferred)
{
	if(e)
	{
		if(_sink)
			_sink->handleEvent(WPE_Close, 0);

		write_log(_sink, LL_ERROR, "[ParserXeleSkt] Error occured while receiving: {}({})", e.message().c_str(), e.value());

		if (!_stopped)
		{
			std::this_thread::sleep_for(std::chrono::seconds(2));
			reconnect();
			return;
		}
	}

	if(_stopped || bytes_transferred<=0)
		return;

	extract_buffer(bytes_transferred);

	_udp_socket->async_receive_from(buffer(_udp_buffer), _udp_ep,
		boost::bind(&ParserXeleSkt::handle_udp_read, this,
			boost::asio::placeholders::error,
			boost::asio::placeholders::bytes_transferred));
}

/*!
 * \brief Xele时间戳处理常量定义
 * 
 * 用于Xele系统特有的时间戳转换和处理
 */
#define XELE_MD_EPOCH 1546272000 // unix timestamp of 2019/1/1 0:0:0 (Xele时间基准点)
#define FOUR_BIT 4				 // 4位偏移量
#define FIVE_HUNDRED_MS 500		 // 500毫秒
#define ZERO_MS 0				 // 0毫秒
#define XELE_MD_MS_MASK 0x000f	 // 毫秒掩码

/*!
 * \brief 解析Xele数据缓冲区
 * \param length 数据长度
 * 
 * \details 解析接收到的Xele数据包，根据消息类型分发处理：
 *          
 *          支持的数据类型：
 *          - MESSAGE_MARKET_DATA: 普通行情数据
 *          - MESSAGE_DEPTH: 深度行情数据（五档盘口）
 *          - MESSAGE_TYPE_HEART_BEAT: 心跳消息
 *          
 *          处理流程：
 *          - 解析数据包头获取版本和消息类型
 *          - 验证Xele协议版本
 *          - 根据类型解析相应的数据结构
 *          - 进行时间戳转换（Xele特有格式）
 *          - 应用价格精度转换
 *          - 更新Tick缓存数据
 *          - 调用回调函数分发数据
 *          - 记录接收统计信息
 * 
 * \note Xele的时间戳格式需要特殊处理，包含毫秒精度
 * \warning 未知的消息类型会被忽略
 */
void ParserXeleSkt::extract_buffer(uint32_t length)
{
	std::size_t len = 0;
	char * data = _udp_buffer.data();
	while (length > len)
	{
		char * data = _udp_buffer.data() + len;
		CXeleShfeMarketHead *mh = reinterpret_cast<CXeleShfeMarketHead *>(data);
		int8_t version = mh->Version;
		if (version != XELE_MD_DATA_VERSION)
			return;

		int8_t type = mh->Type;
		if (type == MESSAGE_MARKET_DATA)
		{
			CXeleShfeMarketData *p = reinterpret_cast<CXeleShfeMarketData *>(data + sizeof(CXeleShfeMarketHead));
			static std::size_t packLen = sizeof(CXeleShfeMarketHead) + sizeof(CXeleShfeMarketData);
			len += packLen;

			//write_log(_sink, LL_DEBUG, "[ParserXeleSkt] Receiving market data, length: {}", packLen);

			WTSTickData* tick = (WTSTickData*)_tick_cache->get(p->InstrumentNo);
			if (tick != NULL)
			{
				uint64_t SnapTime = ((p->SnapDateTime) >> FOUR_BIT) + XELE_MD_EPOCH;
				struct tm *tNow;
				tNow = localtime((time_t*)&SnapTime);
				uint32_t SnapMillisec = (p->SnapDateTime & XELE_MD_MS_MASK) ? FIVE_HUNDRED_MS : ZERO_MS;
				uint32_t actDate = (tNow->tm_year + 1900) * 10000 + (tNow->tm_mon + 1) * 100 + tNow->tm_mday;
				uint32_t actTime = tNow->tm_hour * 10000 + tNow->tm_min * 100 + tNow->tm_sec;
				actTime = actTime * 1000 + SnapMillisec;

				double scale = _price_scales[p->InstrumentNo];
				WTSTickStruct& quote = tick->getTickStruct();
				quote.action_date = actDate;
				quote.action_time = actTime;
				quote.price = p->LastPrice*scale;
				quote.high = std::max(quote.high, p->LastPrice*scale);
				quote.low = std::min(quote.low, p->LastPrice*scale);

				quote.total_volume = p->Volume;
				quote.total_turnover = (double)p->Turnover;
				quote.open_interest = (double)p->OpenInterest;

				quote.volume = 0;
				quote.turn_over = 0;
				quote.diff_interest = 0;

				// 委托价格
				quote.ask_prices[0] = p->AskPrice*scale;
				// 委托价格
				quote.bid_prices[0] = p->BidPrice*scale;
				// 委托量
				quote.ask_qty[0] = p->AskVolume;
				// 委托量
				quote.bid_qty[0] = p->BidVolume;

				if (_sink)
					_sink->handleQuote(tick, 1);

				static uint32_t recv_cnt = 0;
				recv_cnt++;
				if (recv_cnt % _gpsize == 0)
					write_log(_sink, LL_DEBUG, "[ParserXeleSkt] {} ticks received in total", recv_cnt);
			}
		}
		else if (type == MESSAGE_DEPTH)
		{
			CXeleShfeDepthMarketData *p = (CXeleShfeDepthMarketData *)(data + sizeof(CXeleShfeMarketHead));
			static std::size_t packLen = sizeof(CXeleShfeMarketHead) + sizeof(CXeleShfeDepthMarketData);
			len += packLen;

			//write_log(_sink, LL_DEBUG, "[ParserXeleSkt] Receiving depth market data of {}, length: {}", p->InstrumentNo, packLen);

			WTSTickData* tick = (WTSTickData*)_tick_cache->get(p->InstrumentNo);
			if (tick != NULL)
			{
				uint64_t SnapTime = ((p->SnapDateTime) >> FOUR_BIT) + XELE_MD_EPOCH;
				struct tm *tNow;
				tNow = localtime((time_t*)&SnapTime);
				uint32_t SnapMillisec = (p->SnapDateTime & XELE_MD_MS_MASK) ? FIVE_HUNDRED_MS : ZERO_MS;
				uint32_t actDate = (tNow->tm_year + 1900) * 10000 + (tNow->tm_mon + 1) * 100 + tNow->tm_mday;
				uint32_t actTime = tNow->tm_hour * 10000 + tNow->tm_min * 100 + tNow->tm_sec;
				actTime = actTime * 1000 + SnapMillisec;

				double scale = _price_scales[p->InstrumentNo];
				WTSTickStruct& quote = tick->getTickStruct();
				quote.action_date = actDate;
				quote.action_time = actTime;
				quote.price = p->LastPrice*scale;
				quote.high = std::max(quote.high, p->LastPrice*scale);
				quote.low = std::min(quote.low, p->LastPrice*scale);

				quote.total_volume = p->Volume;
				quote.total_turnover = (double)p->Turnover;
				quote.open_interest = (double)p->OpenInterest;

				quote.volume = 0;
				quote.turn_over = 0;
				quote.diff_interest = 0;

				// 五档买卖盘口数据
				quote.bid_prices[0] = p->BidPrice1*scale;
				quote.bid_qty[0] = p->BidVolume1;
				quote.ask_prices[0] = p->AskPrice1*scale;
				quote.ask_qty[0] = p->AskVolume1;

				quote.bid_prices[1] = p->BidPrice2*scale;
				quote.bid_qty[1] = p->BidVolume2;
				quote.ask_prices[1] = p->AskPrice2*scale;
				quote.ask_qty[1] = p->AskVolume2;

				quote.bid_prices[2] = p->BidPrice3*scale;
				quote.bid_qty[2] = p->BidVolume3;
				quote.ask_prices[2] = p->AskPrice3*scale;
				quote.ask_qty[2] = p->AskVolume3;

				quote.bid_prices[3] = p->BidPrice4*scale;
				quote.bid_qty[3] = p->BidVolume4;
				quote.ask_prices[3] = p->AskPrice4*scale;
				quote.ask_qty[3] = p->AskVolume4;

				quote.bid_prices[4] = p->BidPrice5*scale;
				quote.bid_qty[4] = p->BidVolume5;
				quote.ask_prices[4] = p->AskPrice5*scale;
				quote.ask_qty[4] = p->AskVolume5;

				if (_sink)
					_sink->handleQuote(tick, 1);

				static uint32_t recv_cnt = 0;
				recv_cnt++;
				if (recv_cnt % _gpsize == 0)
					write_log(_sink, LL_DEBUG, "[ParserXeleSkt] {} depth ticks received in total", recv_cnt);
			}
		}
		else if (type == MESSAGE_TYPE_HEART_BEAT)
		{
			CXeleShfeHeartBeat *p = (CXeleShfeHeartBeat *)(data + sizeof(CXeleShfeMarketHead));
			static std::size_t packLen = sizeof(CXeleShfeMarketHead) + sizeof(CXeleShfeHeartBeat);
			len += packLen;

			write_log(_sink, LL_DEBUG, "[ParserXeleSkt] Receiving heartbeat, length: {}", packLen);
		}
		else
		{
			write_log(_sink, LL_ERROR, "[ParserXeleSkt] Unknown message type: {}", type);
			return;
		}
	}
}

/*!
 * \brief 处理连接成功事件
 * 
 * \details 连接成功后的回调处理：
 *          - 触发WPE_Connect连接事件
 *          - 触发WPE_Login登录事件
 * 
 * \note Xele系统连接成功后自动视为登录成功
 */
void ParserXeleSkt::doOnConnected()
{
	if(_sink)
	{
		_sink->handleEvent(WPE_Connect, 0);
		_sink->handleEvent(WPE_Login, 0);
	}
}

/*!
 * \brief 处理连接断开事件
 * 
 * \details 连接断开后的回调处理：
 *          - 触发WPE_Close断开事件
 * 
 * \note 通知上层应用连接已断开，可以进行相应的清理工作
 */
void ParserXeleSkt::doOnDisconnected()
{
	if(_sink)
	{
		_sink->handleEvent(WPE_Close, 0);
	}
}